package window;

import bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.junit.Test;

import java.time.Duration;
import java.util.Iterator;

public class Flink_Window_late {

    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> input = env.socketTextStream("node193", 9999).map((MapFunction<String, WaterSensor>) value -> {
            String[] split = value.split(",");
            return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
        });
        //TODO 水位线引入,指定时间字段
        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (element, recordTimestamp) -> element.getTs() * 1000);

        SingleOutputStreamOperator<WaterSensor> result = input.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy)
                .keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //TODO 开窗 ，允许迟到2s，且超过2s后窗口关闭进入侧输出流
                .allowedLateness(Time.seconds(2))
                .sideOutputLateData(new OutputTag<WaterSensor>("lateData") {
                })
//                .reduce((ReduceFunction<WaterSensor>) (value1, value2) -> value2)
                //TODO 自定义窗口计算规则
                .apply(new WindowFunction<WaterSensor, WaterSensor, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<WaterSensor> input, Collector<WaterSensor> out) throws Exception {
                        Iterator<WaterSensor> iterator = input.iterator();
                        while (iterator.hasNext()) {
                            out.collect(iterator.next());
                        }
                    }
                })
                ;
        result.print("result");
        result.getSideOutput(new OutputTag<WaterSensor>("lateData") {
        }).print("late");

        env.execute();
    }
}
